package com.kefu.admin.common.zk.zkleader;

import com.kefu.admin.common.globals.ApplicationConstant;
import com.kefu.admin.common.exception.BaseException;
import org.apache.commons.lang3.StringUtils;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.leader.LeaderSelectorListener;
import org.apache.curator.framework.recipes.leader.LeaderSelectorListenerAdapter;
import org.apache.curator.framework.recipes.locks.InterProcessMutex;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;

import java.io.UnsupportedEncodingException;
import java.util.concurrent.CountDownLatch;

/**
 * Created by TWD on 2018/9/29.
 */
public class ZkSchedulerCoordinator  implements InitializingBean, DisposableBean {

    private static final int DEFAULT_SESSION_TIMEOUT_MS = Integer.getInteger("curator-default-session-timeout", '\uea60').intValue();
    private static final int DEFAULT_CONNECTION_TIMEOUT_MS = Integer.getInteger("curator-default-connection-timeout", 15000).intValue();
    private static Logger logger = LoggerFactory.getLogger(ZkSchedulerCoordinator.class);
    private ApplicationConstant applicationConstant;
    private CountDownLatch latch = null;
    private CuratorFramework client = null;
    private ZKLeaderSelector selector = null;

    public ZkSchedulerCoordinator(ApplicationConstant applicationConstant) {
        this.applicationConstant = applicationConstant;
    }

    public void afterPropertiesSet() throws Exception {
        this.latch = new CountDownLatch(1);
        String zkAddress = this.applicationConstant.zookeeperAddress;
        if(StringUtils.isBlank(zkAddress)) {
            throw new BaseException("配置项app.zookeeper.address不能为空");
        } else {
            String path = ZkUtils.createZkSchedulerLeaderPath(this.applicationConstant.applicationName, this.applicationConstant.profile);
            logger.info("开始连接zookeeper, 进行scheduler leader选举, zookeeperAddress:{}, path:{}", zkAddress, path);
            RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
            synchronized(this) {
                if(this.client == null) {
                    this.client = CuratorFrameworkFactory.builder().connectString(zkAddress).sessionTimeoutMs(DEFAULT_SESSION_TIMEOUT_MS).connectionTimeoutMs(DEFAULT_CONNECTION_TIMEOUT_MS).retryPolicy(retryPolicy).defaultData(this.getDefaultIp()).build();
                    this.client.start();
                }
            }

            LeaderSelectorListener listener = new ZkSchedulerCoordinator.SchedulerLeaderSelector();
            this.selector = new ZKLeaderSelector(this.client, path, listener);
            this.selector.autoRequeue();
            this.selector.start();
        }
    }

    private byte[] getDefaultIp() throws UnsupportedEncodingException {
        String ip = "127.0.0.1";
        return ip.getBytes("UTF-8");
    }

    public void destroy() throws Exception {
        if(this.selector != null) {
            this.selector.close();
        }

        if(this.latch != null) {
            this.latch.countDown();
        }

    }

    public boolean isLeader() {
        return this.selector != null && this.selector.hasLeadership();
    }

    public CuratorFramework getClient() {
        return this.client;
    }

    public InterProcessMutex initXxlJobLock(String jobName) {
        if(StringUtils.isBlank(jobName)) {
            throw new BaseException("jobName不能为空");
        } else {
            return new InterProcessMutex(this.client, ZkUtils.createZkXxlJobLockPath(this.applicationConstant.applicationName, this.applicationConstant.profile, jobName));
        }
    }

    class SchedulerLeaderSelector extends LeaderSelectorListenerAdapter {
        SchedulerLeaderSelector() {
        }

        public void takeLeadership(CuratorFramework client) throws Exception {
            ZkSchedulerCoordinator.logger.info("获取 zookeeper scheduler leader");
            ZkSchedulerCoordinator.this.latch.await();
            ZkSchedulerCoordinator.logger.info("主动放弃 zookeeper scheduler leader");
        }
    }
}
